package job.relevance.anchor_weight;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

import mapred.base.TextMapper;

import org.apache.hadoop.io.Text;

public class RLV_AW_OutLinks_Mapper extends TextMapper {

    private static final String PAGE_URL = "#####PAGE_URL#####";

    private static final String OUT_LINKS = "#####SEGMENTED_OUTLINK#####";

    private static final String FINISH = "#####FINISH_SIGN#####";

    private static final List<String> TYPE = Arrays.asList("ISITEOBODY",
            "ISITEIBODY", "OSITEOBODY", "OSITEIBODY");

    private String last;

    private List<String> buffer;

    private String url;

    @Override
    public void map(Text key, Text value) throws IOException {
        String k = key.toString();
        if (PAGE_URL.equals(last)) {
            url = k;
        } else if (OUT_LINKS.equals(last)) {
            buffer = new ArrayList<String>();
        } else if (FINISH.equals(k)) {
            for (String s : buffer) {
                this.output(new Text(url), new Text("#" + s));
            }
            buffer.clear();
            buffer = null;
            this.incrCounter("Page", "Number", 1);
        } else if (buffer != null) {
            String v = value.toString();
            if (v.contains("\t")) {
                int pos = v.indexOf("\t");
                String tag = v.substring(0, pos);
                buffer.add(k + "\t" + TYPE.indexOf(tag) + v.substring(pos));
            }
        }
        last = k;
    }

    @Override
    public void setUp() {
    }

    @Override
    public void tearDown() throws IOException {
    }

}
